package product.hbase;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.hadoop.hbase.HbaseTemplate;

@Configuration
public class HBaseConfiguration {
	private Logger logger = LoggerFactory.getLogger(this.getClass());

	@Value("${hbase.zookeeper.quorum}")
	private String zookeeperQuorum;

	@Value("${hbase.zookeeper.property.clientPort}")
	private String clientPort;

	@Value("${zookeeper.znode.parent}")
	private String znodeParent;

	@Bean
	public HbaseTemplate hbaseTemplate() {
		org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
		conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
		conf.set("hbase.zookeeper.property.clientPort", clientPort);
		conf.set("zookeeper.znode.parent", znodeParent);
		return new HbaseTemplate(conf);
	}

	public Connection connection() {
		try {
			org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
			conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
			conf.set("hbase.zookeeper.property.clientPort", clientPort);
			conf.set("zookeeper.znode.parent", znodeParent);

			ExecutorService pool = Executors.newScheduledThreadPool(20); //设置hbase连接池
			return ConnectionFactory.createConnection(conf, pool);
		} catch (IOException e) {
			logger.error("HbaseUtils实例初始化失败！错误信息为：" + e.getMessage(), e);
		}
		return null;
	}
}